package com.atguigu.day02

import org.apache.flink.streaming.api.functions.co.CoMapFunction
import org.apache.flink.streaming.api.scala._

object CoMapExample {
  def main(args: Array[String]): Unit = {
    val env:StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

    val stream1: DataStream[(String, Int)] = env.fromElements(("wangchenglong", 21),
      ("wangrong", 11))
    val stream2: DataStream[(String, Int)] = env.fromElements(("wangchenglong", 100),
      ("wangrong", 50))

    //直接来连接两条流没有意义
    //必须把相同的key连接起来
    val connect: ConnectedStreams[(String, Int), (String, Int)] = stream1.keyBy(_._1)
      .connect(stream2.keyBy(_._1))

    //方法二
//   val connect: ConnectedStreams[(String, Int), (String, Int)] =  stream1.connect(stream2).keyBy(0,0)

    connect.map(new CoMyMapFunction).print()
    env.execute()
  }

  class CoMyMapFunction extends CoMapFunction[(String, Int),(String, Int),String]{
    override def map1(value: (String, Int)): String = {
      value._1+"的体重是"+value._2+"斤"
    }

    override def map2(value: (String, Int)): String = {
      value._1+"的年龄是"+value._2+"岁"
    }
  }
}
